package com.github.ompc.carrying.client.worker;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;

import com.github.ompc.carrying.client.array.DataConsumerArrayManager;
import com.github.ompc.carrying.client.persistence.CarryingDataPersistenceDao;
import com.github.ompc.carrying.client.util.BytesReverseUtil;
import com.github.ompc.carrying.common.domain.Row;
import com.github.ompc.carrying.common.service.CarryingService;

/**
 * 用于数据消费的线程池管理类
 * 
 * @author admin
 *
 */
public class CarryingDataWorkerManager implements InitializingBean {

	private final Logger logger = LoggerFactory.getLogger(getClass());

	private int carryingDataThreadNum = 12;
	
	private ExecutorService carryingWorkers = Executors.newFixedThreadPool(
			carryingDataThreadNum,new CarryingDataClientThreadFactory());

	private CarryingDataPersistenceDao mappedCarryingDataPersistenceDao;
	private DataConsumerArrayManager dataConsumerArrayManager;
	private CarryingService carryingService;

	private CountDownLatch latch = new CountDownLatch(carryingDataThreadNum);
	
	@Override
	public void afterPropertiesSet() throws Exception {
		
		Thread.sleep(1000*3);
		
		for (int i = 0; i < carryingDataThreadNum; i++) {
			carryingWorkers.execute(new CarryingDataWorker(i,latch));
			logger.info("info@CarryingDataWorkerManager afterPropertiesSet carryingDataThread{} start",i);
		}
		
		latch.await();
		
		carryingWorkers.shutdown();

		mappedCarryingDataPersistenceDao.persistenceData();

	}

	/**
	 * 内部类，向server请求数据
	 */
	class CarryingDataWorker implements Runnable {
		private Row row = null;
		private boolean running = true;
		// 线程标识
		private final int num;
		// 消息计数器
		private int counter;
		// 向服务器请求数据失败，重试次数
		private boolean reTry = false;
		private CountDownLatch latch;
		CarryingDataWorker(int num,CountDownLatch latch){
			this.num = num;
			this.latch = latch;
		}

		/**
		 * token生成策略 1.线程标识+移位
		 */
		@Override
		public void run() {
			while (running) {
				try {
					row = carryingService.carry(counter, reTry);
					logger.info("info@CarryingDataWorker lineNum={};data={}",new Object[]{row.getLineNum(),new String(row.getData())});
					reTry = false;
					counter++;
					counter <<= 8;
					counter = counter < 0 ? 0 : (counter + num);
				} catch (Exception e) {
					logger.error(
							"error@CarryingDataWorkerManager CarryingDataWorker,num={};reTry={}",
							new Object[] { num, reTry},e);
					reTry = true;
					continue;
				}

				if (null == row) {
					running = false;
					continue;
				}
				row.setData(BytesReverseUtil.reverse(row.getData()));
				dataConsumerArrayManager.put(row);
			}
		
			latch.countDown();

		}
	}

	public void setMappedCarryingDataPersistenceDao(
			CarryingDataPersistenceDao mappedCarryingDataPersistenceDao) {
		this.mappedCarryingDataPersistenceDao = mappedCarryingDataPersistenceDao;
	}

	public void setDataConsumerArrayManager(
			DataConsumerArrayManager dataConsumerArrayManager) {
		this.dataConsumerArrayManager = dataConsumerArrayManager;
	}
	
	public void setCarryingService(CarryingService carryingService) {
		this.carryingService = carryingService;
	}

	public void setCarryingDataThreadNum(int carryingDataThreadNum) {
		this.carryingDataThreadNum = carryingDataThreadNum;
	}
}
